…cate writes

Under TP-only configs (e.g. TP=2 CP=1), every rank in the TP group
was calling ``TQWorkerMixin._write_back`` and racing to write the
same per-sample keys (``prev_logprobs`` etc.). On the simple
backend the second writer's bytes silently overwrote the first
(``last-write-wins`` on a Python dict — benign because the data is
identical post-all-reduce). On the mooncake_cpu backend the
``MooncakeStore`` master rejected the second writer's
``BatchPutEnd`` with ``ILLEGAL_CLIENT`` (-601) because the metadata
``client_id`` was set to the first writer's UUID — the recipe
crashed at the first ``kv_batch_put`` of the offending step.

The existing leader check ``_is_replica_leader`` correctly returns
False for non-leaders, but only when ``_get_replica_group`` returns
a non-None group. Subclasses gate ``_get_replica_group`` on
``CP > 1`` as a fetch-path optimization (the docstring explicitly
calls out "matches the qwen3-mcore TP=2 baseline"). That gate
incorrectly disables the leader check on the write-back path too:
``CP=1 ⇒ replica_group is None ⇒ _is_replica_leader → True for
every rank``.

Split the write-back leader check from the replica-group machinery:

- ``TQWorkerMixin._is_writeback_leader``: default delegates to
  ``_is_replica_leader`` (preserves behavior for workers with no
  parallelism).
- ``MegatronPolicyWorkerImpl._is_writeback_leader``: override gates on
  ``(tp_rank, cp_rank, pp_rank) == (0, 0, 0)`` via mcore
  ``parallel_state`` — unconditional, no CP gate.
- ``DTensorPolicyWorkerV2Impl._is_writeback_leader``: same idea but
  using ``device_mesh["cp"].get_local_rank()`` /
  ``device_mesh["tp"].get_local_rank()``.
- ``_write_back`` switched from ``_is_replica_leader`` to
  ``_is_writeback_leader``.

Correctness: simple backend's ``last-write-wins`` already proves the
data is identical across TP siblings (DSv3 32n8g TP=32 simple passes
its closing ``check_metrics.py`` with the same multi-write pattern;
gating to leader-only is semantically equivalent). Mooncake's race is
eliminated because exactly one client now writes each key.

Perf: ``tp_size - 1`` redundant ``kv_batch_put`` calls per training
step are now skipped on every backend, not just mooncake_cpu.

Verified by JOBID 11758259 (1n8g megatron TP=2 + temp/top-p/top-k
sampling on mooncake_cpu) — past Step 11/500 with no -601, whereas
every prior attempt of this recipe crashed at Step 1 within ~5 min.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>

DataPlaneClient stays a pure transport. Rollout-shape concerns (key
layout, batch geometry) move to the producer.

* New codec.pack_jagged_fields() owns the single wire-layout transform
  (jagged pack + np.ndarray(dtype=object) passthrough). Previously
  duplicated across kv_first_write, write_columns, and
  worker_mixin._write_back.
* kv_first_write moves to data_plane/column_io.py next to write_columns
  and takes pre-minted keys (no longer (uids, n_gen)). Rollout actor
  builds the keys inline at the call site, matching how verl's
  AgentLoopWorkerTQ._agent_loop_postprocess draws the same line.
* worker_mixin._write_back collapses to write_columns(...).
* Unit tests updated for the new keys= signature.

Net: sync_rollout_actor.py -95 LOC; client signature unchanged.

Per yuki-97 PR review (#2).

Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>

* Remove test_materialize_rejects_non_tensor_leaves: codec.materialize
  now deliberately accepts NonTensorData (commit c3ac42342, simple-
  backend wire-strip survival). The "wire is tensors only" assertion
  predated that fix and contradicts current behavior.
* Remove test_no_data_plane_in_master_config: TODO gate that fires
  until legacy grpo.py is retired; tracked separately.
* Rewrite test_codec_wire_stripped.py to cover the production decode
  paths via:
    - direct unit coverage of unwrap_wire_stripped_payload (per-item
      helper; doesn't need a NonTensorStack construction);
    - end-to-end materialize coverage via patch.object(stack, "tolist")
      to simulate the wire-stripped state (tensordict>=0.12.2 rejects
      NonTensorStack(TensorDict({}, batch_size=[])) at construction).
  Wire round-trip e2e for object columns is already covered by
  tests/data_plane/functional/test_tq_lifecycle.py.

Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>

* grpo_sync.py: remove unused batch_cache = None (leftover from
  grpo.py-style dynamic sampling; grpo_sync threads survivors through
  pending_meta / pending_slice).
* TQPolicy: rename _dp_client -> dp_client and _tq_partition_id ->
  tq_partition_id. They are read from grpo_sync.py in 7 places, so
  the underscore prefix was misleading. Constructor kwarg
  tq_partition_id already matched the new attribute name.
* Update README + data_plane_api_lifecycle docs example snippets.

Per yuki-97 PR review (#3, #4).

Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>

…ema constant

Replaces the inline 6-field exclusion tuple in grpo_sync.py with
DP_CALIB_EXCLUDED_FIELDS in data_plane/schema.py, derived from
DP_TRAIN_FIELDS - {input_ids, input_lengths}. A new column added to
DP_TRAIN_FIELDS is now excluded-by-default from KV-scale calibration;
opting it in requires editing the private base set explicitly.

Multimodal extras (pixel_values, image_grid_thw, etc.) pass through
unchanged because they are not in DP_TRAIN_FIELDS.

Per yuki-97 PR review (#5).

Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>

Silent over-fetch was possible when callers omitted select_fields: the
noop adapter fetched every registered field via set intersection; the
TQ adapter forwarded None to the backend. Bulk schemas are wide and
fetching everything is the most expensive shape the wire can take.

select_fields is now a required list[str] on DataPlaneClient.kv_batch_get
and all concrete implementations. Callers must name what they read;
fetch-all is still possible by passing list(meta.fields) explicitly.

Also: worker_mixin internal call sites use list(meta.fields) directly
(fail-loud TypeError if meta.fields is None, rather than silently
producing an empty TensorDict).

Per yuki-97 PR review (#6).

Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>

…ifecycle

Remove three actor-level wrappers (finish_generation, get_logger_metrics,
clear_logger_metrics) that had zero external callers. The actor's internal
code already calls self.policy_generation.{...} directly at the right
points inside rollout_to_tq; the wrappers added indirection without value.

Rewrite the rollout_to_tq docstring to list all six steps bundled into
the single Ray RPC (reset metrics -> rollout -> flatten -> TQ put ->
release GPU -> capture metrics), making the lifecycle visible without
having to read the method body.

Per yuki-97 PR review (#7, #8).

Signed-off-by: Zhiyu Li <zhiyul@NVIDIA.com>